package com.atguigu.flink.datastramapi.transform;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created by Smexy on 2023/2/24
 *
 *  OperatorChain:  算子链 或算子合并。
 *          把多个算子合并为一个算子。
 *
 *      合并的条件一：  两个算子必须是forward分发规则。
 *                      细分：  ①上下游并行度必须相等
 *                             ②必须是1:1
 *                                  不能是1:N  shuffle,rescale,reblance,keyBy
 *                                  不能是N:1  global
 *
 *
 *      合并的条件二： 算子的SlotSharingGroup 必须相同。
 *                      SlotSharingGroup 是一个标记，标识了，算子是否合得来！
 *                      SlotSharingGroup 会向下传递，默认所有的算子的SlotSharingGroup是一样的。
 *
 *       --------------
 *          默认开启，将算子合并后，可以减少上下游数据的传输。
 *
 *
 */
public class Demo2_OperatorChain
{
    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        //全局禁止
        //env.disableOperatorChaining();

        DataStreamSource<String> ds = env.socketTextStream("hadoop103", 8888);  //default

        ds.map(x->x).name("map1").setParallelism(1).slotSharingGroup("a")  //a
          .map(x->x).name("map2")  //a
          .map(x->x).name("map3").slotSharingGroup("b")  //b
          //禁止和前后的算子进行合并
          .map(x->x).name("map4")
          //从这个算子开启一个新算子链。 不和前面的合并，可以和后面的合并
          .startNewChain()//.disableChaining()   //b
          .print();  //b

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
